package org.apache.hadoop.contrib.mfs;

import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

public class MetadataManager {
	
	private static final ConcurrentHashMap<String,FileInfo> metadatas = new ConcurrentHashMap<String,FileInfo>();
        private static Chunk[] chunks;
        private static MetadataManager instance = new MetadataManager();
        static {
            
        }

        private static long chunkSize = -1;
        private static long totalStorageSize;
        private static long freeStorageSize;

        // unique numbering counter needs for the future features (chunk extending)
        private static AtomicLong counter;
	
	private MetadataManager(){
            
            setTotalStorageSize( Long.parseLong(MfsConfiguration.config.get("mfs.storage.totalSize").toString()) );
            setChunkSize( Long.valueOf(MfsConfiguration.config.get("mfs.storage.chunkSize").toString()) );

            counter = new AtomicLong(0L);
            // Initialize chunks
            int chunkCount = (int) (getTotalStorageSize() / getChunkSize());
            chunks = new Chunk[chunkCount];
            for(int i=0; i<chunks.length; i++){
                chunks[i] = new Chunk();
                chunks[i].seq = counter.getAndIncrement();
            }
            metadatas.clear();
        }
	public static MetadataManager getInstance(){
            return instance;
	}
	
	public void add(FileInfo fileInfo){
            String path = fileInfo.getAbsolutePath();
            if(metadatas.containsKey(path)){
                Logger.getLogger("MetadataManager").log(Level.WARNING, "Already exists");
                return;
            }

            fileInfo.updateTimestamp();
            metadatas.put(path, fileInfo);

            setFreeStorageSize(getTotalStorageSize() - fileInfo.getChunkList().size() * chunkSize);
	}
	
	public void update(FileInfo modifiedFile){
            long size = 0L;

            synchronized(metadatas){
                FileInfo temp = metadatas.get(modifiedFile.getPath());
                size -= temp.getChunkList().size() * chunkSize;

                temp = modifiedFile;
                temp.updateTimestamp();
                
                size += modifiedFile.getChunkList().size() * chunkSize;
            }
            setFreeStorageSize(freeStorageSize + size);
	}
	
	public void delete(String path) throws FileNotFoundException {
            List<Chunk> chunks = find(path).getChunkList();
            freeChunks(chunks);
            
            metadatas.remove(path);
	}

        public FileInfo find(String path) throws FileNotFoundException{
            FileInfo temp = metadatas.get(path);
            if(temp==null){
                throw new FileNotFoundException();
            }else{
                return temp;
            }
        }
	
	public FileInfo[] list(String path){
            ArrayList<FileInfo> list = new ArrayList();

            Enumeration<String> e =  metadatas.keys();
            while(e.hasMoreElements()){
                String key = e.nextElement();
                if(key.startsWith(path)){
                    list.add(metadatas.get(key));
                }
            }

            FileInfo[] temp = new FileInfo[list.size()];
            return list.toArray(temp);
	}

        /*
         * In this process, this method does not make a possession of selected chunks.
         * So, IT MAY OCCUR PROBLEMS if this method calls concurrently.
         */
        public List<Chunk> allocateChunks(long size) throws InsufficientSpaceException{
            int chunkCount = (size <= getChunkSize()) ? 1 : (int)Math.ceil(size / getChunkSize());

            ArrayList<Chunk> list = new ArrayList<Chunk>(chunkCount);

            // 분포를 확인하여 검색 범위를 다르게 할수 있지만, 일단 바이너리 검색...
            // MetadataManager가 하나이므로 이용할 청크를 선택하였더라도 청크플래그에 기록하지않음
            for(int i=0; i<chunks.length; i++){
                if((chunks[i].flag & Chunk.ALLOCATED) == Chunk.NOT_ALLOCATED ){
                    list.add(chunks[i]);
                    if(list.size() == chunkCount) break;
                }
            }
            if(list.size() != chunkCount){
                throw new InsufficientSpaceException("Failed to allocate Chunks");
            }else{
                return list;
            }
        }

        public void markChunks(List<Chunk> chunks){
            Iterator<Chunk> iter = chunks.iterator();
            while(iter.hasNext()){
                iter.next().flag |= (Chunk.UPDATING | Chunk.ALLOCATED);
            }
        }
        public void unmarkChunks(List<Chunk> chunks){
            Iterator<Chunk> iter = chunks.iterator();
            while(iter.hasNext()){
                iter.next().flag &= ~Chunk.UPDATING;
            }
        }

        public void freeChunks(List<Chunk> chunks){
            Iterator<Chunk> iter = chunks.iterator();
            while(iter.hasNext()){
                iter.next().flag &= ~Chunk.ALLOCATED;
            }
        }
        /*
         * When Storage size has been changed, total available space and free space both will be renewed.
         * and, Chunk Table which has Sequence Number, Block Status Info also will be.
         * Note That PREVIOUS Metadatas WILL BE LOST. currently NOT SUPPORT.
         */
        private void setTotalStorageSize(long totalStorageSize){
            MetadataManager.totalStorageSize = totalStorageSize;
            MetadataManager.freeStorageSize = totalStorageSize;
        }

        public static long getTotalStorageSize(){
            return MetadataManager.totalStorageSize;
        }
        private void setFreeStorageSize(long freeStorageSize){
            if(freeStorageSize > totalStorageSize){
                Logger.getLogger("MetadataManager").log(Level.WARNING, "FreeStorageSize exceeds TotalStorageSize");
                return;
            }
            MetadataManager.freeStorageSize = freeStorageSize;
        }
        public static long getFreeStorageSize(){
            return freeStorageSize;
        }

        public static long getChunkSize(){
            return Long.parseLong(MfsConfiguration.config.getProperty("mfs.storage.chunkSize"));
        }
        private void setChunkSize(long size){
            if(size > MetadataManager.getTotalStorageSize()){
                Logger.getLogger("MetadataManager").log(Level.WARNING, "ChunkSize exceeds TotalStorageSize");
                return;
            }
            chunkSize = size;

        }

        public boolean isAvailableStorage(){
            return (MetadataManager.getTotalStorageSize() / chunkSize) > 0 ? true : false;
        }
        
}